// Description: 描述
package mq

import (
	"encoding/json"
	"fmt"
	"os/user"
	"strings"
	"time"

	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/core/service"
	"github.com/zeromicro/go-zero/core/stores/cache"
	"github.com/zeromicro/go-zero/core/stringx"
	"github.com/zeromicro/go-zero/core/threading"

	"gitlab.vspncd.com/mic/common/mq/internal/dq"
	"gitlab.vspncd.com/mic/common/tools"
)

type (
	DelayConsume func(key string, body []byte)

	DelayMq interface {
		// 设置
		SetUp(adds []string, tube string, redisConfig cache.NodeConf)
		// 注册消费handle
		RegConsumer(key string, consume DelayConsume)
		// 启动客户端
		Start()
		// 停止客户端
		Stop()
		// 发送定时消息
		At(key string, v interface{}, at time.Time) (string, error)
		// 发送延迟消息
		Delay(key string, v interface{}, delay time.Duration) (string, error)
		// 撤销消息
		Revoke(id string) error
	}

	defaultDelayMq struct {
		mode       string // 启动模式
		producer   dq.Producer
		consumer   dq.Consumer
		consumes   map[string]DelayConsume
		consumeSrv *service.ServiceGroup
	}

	delayMsg struct {
		Key  string `json:"key"`
		Body []byte `json:"body"`
	}
)

func NewDelayMq(mode ...string) DelayMq {
	target := &defaultDelayMq{
		consumes: map[string]DelayConsume{},
	}

	if len(mode) > 0 {
		target.mode = mode[0]
	} else {
		target.mode = "pro"
	}

	return target
}

// tube 用于区分是那个服务端
func (k *defaultDelayMq) SetUp(adds []string, tube string, redisConfig cache.NodeConf) {
	group := fmt.Sprintf("%s%s", k.topicPrefix(), tube)

	// 消费端
	dqCfg := dq.Conf{}
	dqCfg.Redis.Host = redisConfig.Host
	dqCfg.Redis.Type = redisConfig.Type
	dqCfg.Redis.Pass = redisConfig.Pass

	for _, v := range adds {
		dqCfg.Beanstalks = append(dqCfg.Beanstalks, dq.Beanstalk{
			Endpoint: v,
			Tube:     group,
		})
	}

	// 生产端
	k.producer = dq.NewProducer(dqCfg.Beanstalks)
	// 消费端
	k.consumer = dq.NewConsumer(dqCfg)
}

func (k *defaultDelayMq) RegConsumer(key string, consume DelayConsume) {
	k.consumes[key] = consume
}

func (k *defaultDelayMq) Start() {
	if k.consumer == nil {
		return
	}

	k.consumeSrv = k.consumer.Consume(k.consume)
	threading.GoSafe(func() {
		k.consumeSrv.Start()
	})
}

func (k *defaultDelayMq) consume(body []byte) {
	msg := &delayMsg{}
	err := json.Unmarshal(body, msg)
	if err != nil {
		logx.Error("消费消息失败", err)
		return
	}

	if c, ok := k.consumes[msg.Key]; ok {
		c(msg.Key, msg.Body)
	} else {
		logx.Error("无效的消息", msg.Key, string(msg.Body))
		return
	}
}

func (k *defaultDelayMq) Stop() {
	if k.producer != nil {
		_ = k.producer.Close()
	}

	if k.consumeSrv != nil {
		k.consumeSrv.Stop()
	}
}

func (k *defaultDelayMq) At(key string, v interface{}, at time.Time) (string, error) {
	body, err := json.Marshal(v)
	if err != nil {
		return "", err
	}

	m, _ := json.Marshal(&delayMsg{
		Key:  key,
		Body: body,
	})

	taskId := ""
	err = tools.DoWithRetry(func() (e error) {
		taskId, e = k.producer.At(m, at)
		if e != nil {
			return e
		}
		return nil
	})
	if err != nil {
		logx.Error("send msg error", err)
		return "", err
	}

	return taskId, nil
}

func (k *defaultDelayMq) Delay(key string, v interface{}, delay time.Duration) (string, error) {
	body, err := json.Marshal(v)
	if err != nil {
		return "", err
	}

	m, _ := json.Marshal(&delayMsg{
		Key:  key,
		Body: body,
	})

	taskId := ""
	err = tools.DoWithRetry(func() (e error) {
		taskId, e = k.producer.Delay(m, delay)
		if e != nil {
			return e
		}
		return nil
	})
	if err != nil {
		logx.Error("send msg error", err)
		return "", err
	}

	return taskId, nil
}

func (k *defaultDelayMq) Revoke(id string) error {
	return tools.DoWithRetry(func() error {
		return k.producer.Revoke(id)
	})
}

// 通过在topic前面加前缀，来隔离不同环境的消费乱串的问题
func (k *defaultDelayMq) topicPrefix() string {
	prefix := ""
	switch k.mode {
	case "dev":
		u, err := user.Current()
		if err != nil {
			prefix = stringx.Rand() + "_"
		} else {
			prefix = strings.Split(u.Username, "\\")[0] + "_"
			prefix = strings.Split(prefix, "-")[0] + "_"
		}
	case "test":
		prefix = "test_"
	case "pre":
		prefix = "pre_"
	case "pro":
	default:
		panic("环境变量有问题")
	}
	return prefix
}
